package com.example.service.impl;

import com.example.util.ConvertUtil;
import com.example.vo.BaseVo;
import com.example.vo.PageResult;
import com.example.vo.QuerySpecification;
import com.example.repository.BaseRepository;
import com.example.service.BaseReactiveService;
import org.dozer.Mapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Sort;
import org.springframework.data.jpa.domain.Specification;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.Assert;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;

import java.lang.reflect.ParameterizedType;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.Callable;

@Service
@Transactional
public abstract class BaseReactiveServiceImpl<T> implements BaseReactiveService<T> {

    @Autowired
    Scheduler scheduler;

    @Autowired
    Mapper mapper;

    abstract protected BaseRepository<T> getRepository();

    @Override
    public <R> Flux<PageResult<R>> findAll(BaseVo queryVo) {
        Specification<T> spec = QuerySpecification.buildQueryParam(queryVo.buildQueryParams());
        PageRequest pageRequest = queryVo.buildPageRequest();
        Page<T> page = getRepository().findAll(spec, pageRequest);
        PageResult<R> result = new PageResult<R>(page.getNumber(),
                (int) page.getTotalElements(), ConvertUtil.convertObject(mapper, page.getContent(), getRClass()));
        return Flux.just(result).subscribeOn(scheduler);
    }

    @Override
    public <R> Flux<PageResult<R>> findAllBySort(BaseVo queryVo, Sort sort) {
        Specification<T> spec = QuerySpecification.buildQueryParam(queryVo.buildQueryParams());
        List<T> list = getRepository().findAll(spec, sort);
        return Flux
                .just(new PageResult<R>(list.size(), ConvertUtil.convertObject(mapper, list, getRClass())))
                .subscribeOn(scheduler);
    }

    @Override
    public <S extends T, R> Mono<R> save(S entity) {
        mapper.map(getRepository().save(entity), getRClass());
        return async(() -> (R) mapper.map(getRepository().save(entity), getRClass()));
    }

    @Override
    public <S extends T, R> Flux<R> saveAll(Iterable<S> entities) {
        Assert.notNull(entities, "Entities must not be null!");
        return async(ConvertUtil.convertObject(mapper, getRepository().saveAll(entities), getRClass()));
    }

    @Override
    public <R> Mono<R> findById(String id) {
        Assert.notNull(id, "The given id must not be null!");
        Optional<T> optional = getRepository().findById(id);
        return async(getRepository().findById(id).isPresent() ? (Callable<R>) mapper.map(optional.get(), getRClass()) : null);
    }

    @Override
    public Mono<Boolean> existsById(String id) {
        Assert.notNull(id, "The given id must not be null!");
        return Mono.fromCallable(() -> getRepository().existsById(id)).subscribeOn(scheduler);
    }

    @Override
    public <R> Flux<R> findAll() {
        return async(ConvertUtil.convertObject(mapper, getRepository().findAll(), getRClass()));
    }

    @Override
    public <R> Flux<R> findAllById(Iterable<String> ids) {
        return async(ConvertUtil.convertObject(mapper, getRepository().findAllById(ids), getRClass()));
    }

    @Override
    public Mono<Long> count() {
        return Mono.fromCallable(() -> getRepository().count()).subscribeOn(scheduler);
    }

    @Override
    public Mono<Void> deleteById(String id) {
        Assert.notNull(id, "The given id must not be null!");
        getRepository().deleteById(id);
        return Mono.empty();
    }

    @Override
    public Mono<Void> delete(T entity) {
        getRepository().delete(entity);
        return Mono.empty();
    }

    @Override
    public Mono<Void> deleteAll(Iterable<? extends T> entities) {
        Assert.notNull(entities, "Entities must not be null!");
        getRepository().deleteAll(entities);
        return Mono.empty();
    }

    @Override
    public Mono<Void> deleteAll() {
        getRepository().deleteAll();
        return Mono.empty();
    }

    public <O> Mono<O> async(Callable<O> callable) {
        if (callable == null) {
            return Mono.empty();
        }
        return Mono.fromCallable(callable).subscribeOn(scheduler);
    }

    public <O> Flux<O> async(Iterable<O> entities) {
        return Flux.defer(() -> Flux.fromIterable(entities)).subscribeOn(scheduler);
    }

    public <R> Class<R> getRClass() {
        return (Class<R>) ((ParameterizedType) getClass().getGenericSuperclass()).getActualTypeArguments()[0];
    }
}
